#define FD_SETSIZE 4096

#include <cassert>
#include <ctime>
#include <iostream>

#ifndef WIN32
#include <signal.h>
#endif

#include "miniutil/os_compatible.h"
#include "miniutil/socket.h"
#include "miniutil/socketcmn.h"
#include "miniutil/buffer_alloc.h"
#include "miniutil/threadpool.h"
#include "miniutil/socketmul_select.h"

/*
	SocketMulplexService accept socket instance and try to check if it can read/write/exception
	it will not create socket, close socket, and so on
	Invoker should control all these resources. 

	must call reset_socket_event to continue;
*/
namespace miniutil { namespace multinet {

    SocketMulplexService_Select::SocketMulplexService_Select()
    {
        pwritedfs = NULL;
        preaddfs = NULL;
        pexcepdfs = NULL;
    }

    SocketMulplexService_Select::~SocketMulplexService_Select()
    {
        if(pwritedfs != NULL)
            delete (fd_set*)pwritedfs;
        if(preaddfs != NULL)
            delete (fd_set*)preaddfs;
        if(pexcepdfs != NULL)
            delete (fd_set*)pexcepdfs;
    }

	//---------------------------------------------------------------
	void SocketMulplexService_Select::run()
	{
	
		timeval tv;
		tv.tv_sec = 0;
		tv.tv_usec = 1000*10;
		
		stopstus =  false;

        if(pwritedfs != NULL)
            delete (fd_set*)pwritedfs;
        if(preaddfs != NULL)
            delete (fd_set*)preaddfs;
        if(pexcepdfs != NULL)
            delete (fd_set*)pexcepdfs;

        pwritedfs = new fd_set();
        preaddfs = new fd_set();
        pexcepdfs = new fd_set();

		while(true)
		{

			if(stopcmd)
			{
				stopcmd = false;
				break;
			}

			do
			{
				miniutil::auto_lock kk(&msockmtx);

				SOCKET max_socket = init_fdset();
                if(max_socket == 0x00)
                {
                    thread::sleep(10);
                    continue;
                }
				
				int ret = select((int)(max_socket+1), (fd_set*)preaddfs, (fd_set*)pwritedfs, (fd_set*)pexcepdfs, &tv);
				if ( ret < 0 )
				{
		#ifdef WIN32
					int sret = WSAGetLastError();
                    M_DEBUG("select error:%d\n", sret);
		#endif
				    thread::sleep( 10 );
					continue;
				}
				else if ( ret == 0 ) 
				{
				    thread::sleep( 10 );
					continue;
				}

				time_t now;
				time(&now);
				if(now > mprechktime)
				{
                    //M_DEBUG("Restore QUota:%d\n", max_uquota);
					left_dquota = max_dquota;
					left_uquota = max_uquota;
					mprechktime = now;
				}
				
				if(left_dquota > 0)
					this->do_readable( );
				if(left_uquota > 0)
					this->do_writable( );

				do_exception();
			}while(0);

			miniutil::thread::sleep(1);
		}
		stopstus = true;
	}

	SOCKET SocketMulplexService_Select::init_fdset()
	{
		SOCKET max_socket = 0;
		FD_ZERO((fd_set*)pwritedfs);
		FD_ZERO((fd_set*)preaddfs);
		FD_ZERO((fd_set*)pexcepdfs);	
		

		tcpchecksocs.clear();
		udpchecksocs.clear();
		do  //add all socket to ...
		{
			miniutil::auto_lock kk(&msockmtx);
			std::map<SOCKET, MonTcpSocketInstance>::iterator iter_vsmt;
			for(iter_vsmt = malltcpsockets.begin(); iter_vsmt != malltcpsockets.end();)
			{
				if(iter_vsmt->second.closeflag)
				{
					malltcpsockets.erase(iter_vsmt++);
					continue;
				}
				MonTcpSocketInstance &ains = iter_vsmt->second;

				//if(iter_vsm->second.socevent != EVT_NONE)
				//{
				//	iter_vsm ++;
				//	continue;
				//}

				tcpchecksocs.push_back(iter_vsmt->first);
				if((ains.waitevt & EVT_INCOMING ) != 0x00 &&
					(ains.socevent & EVT_INCOMING) == 0x00)
                {
					FD_SET(iter_vsmt->first, (fd_set*)preaddfs);
				    if(iter_vsmt->first > max_socket)
					    max_socket = iter_vsmt->first;
                }

				if((ains.waitevt & EVT_READ_READY) != 0x00 &&
					ains.recvret > 0 &&
					ains.recvbuf.have_space(1024))
				{
					//if(iter_vsm->second.recvlen > 0) //still data there
					//	iter_vsm->second.socevent |= EVT_READ_READY;
					//else
						FD_SET(iter_vsmt->first, (fd_set*)preaddfs);
				    if(iter_vsmt->first > max_socket)
					    max_socket = iter_vsmt->first;
				}
				if((ains.waitevt & EVT_CONNECTED) != 0x00 &&
					(ains.socevent & EVT_CONNECTED) == 0x00)
                {
					FD_SET(iter_vsmt->first, (fd_set*)pwritedfs);
                    FD_SET(iter_vsmt->first, (fd_set*)pexcepdfs);
				    if(iter_vsmt->first > max_socket)
					    max_socket = iter_vsmt->first;
                }

				if(/*(iter_vsmt->second.waitevt & EVT_WRITE_READY) != 0x00 &&*/
					ains.sendret > 0 &&
					ains.send_queue.size() > 0)
                {
                    //M_DEBUG("FD_SETWRITE %d\n", ains.soc);
					FD_SET(iter_vsmt->first, (fd_set*)pwritedfs);
				    if(iter_vsmt->first > max_socket)
					    max_socket = iter_vsmt->first;
                }

				if((ains.waitevt & EVT_EXCEPTION) != 0x00 &&
					(ains.socevent & EVT_EXCEPTION) == 0x00)
                {
					FD_SET(iter_vsmt->first, (fd_set*)pexcepdfs);
				    if(iter_vsmt->first > max_socket)
					    max_socket = iter_vsmt->first;
                }

				iter_vsmt ++;
			}

			std::map<SOCKET, MonUdpSocketInstance>::iterator iter_msmu;
			for(iter_msmu = malludpsockets.begin(); iter_msmu != malludpsockets.end();)
			{
				if(iter_msmu->second.closeflag)
				{
					malludpsockets.erase(iter_msmu++);
					continue;
				}
				//low eff
				if(iter_msmu->second.vsendbuf.size() > 0)
				{
					FD_SET(iter_vsmt->first, (fd_set*)pwritedfs);
				}
				FD_SET(iter_msmu->first, (fd_set*)preaddfs);
				udpchecksocs.push_back(iter_msmu->first);
				iter_msmu ++;
			}

		}while(0);
		return max_socket;
	}

	void SocketMulplexService_Select::do_readable()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], (fd_set*)preaddfs))
			{
				if((ins.waitevt & EVT_INCOMING) != 0x00)
					ins.socevent |= EVT_INCOMING;
				else
				{
					std::pair<char*, unsigned int> writepos;
					ins.recvbuf.getwritepos(writepos);

                    if(left_dquota <= 0)
                        break;
                    if((int)writepos.second > left_dquota)
                        writepos.second = left_dquota;

					assert(writepos.second > 0);
					{
						ins.recvret = ::recv(ins.soc, writepos.first,writepos.second, 0);
						if(ins.recvret > 0)
						{
                            M_DEBUG("recvret %d\n", ins.recvret);
							ins.recvbuf.more_content(ins.recvret);
							left_dquota -= ins.recvret;
							if(left_dquota <= 0)
								break;
						}
						else
						{
							//int ret = miniutil::socket::getLastError();
                            int ret = miniutil::socketcmn::check_cursoc_error();
                            M_DEBUG("recvret < 0 ret:%d, err:%s\n", ret, miniutil::socketcmn::get_soc_errorstr(ret).c_str());
						}
					}
					ins.socevent |= EVT_READ_READY;
				}
			}
		}
		for(unsigned int i=0; i< udpchecksocs.size(); i++)
		{
			socklen_t nmln = sizeof(struct sockaddr);
			MonUdpSocketInstance &ins = malludpsockets[udpchecksocs[i]];
			if(FD_ISSET(udpchecksocs[i], (fd_set*)preaddfs))
			{
				MonUdpSocketInstance::UDPMSG* pmsg = new MonUdpSocketInstance::UDPMSG();
				pmsg->mlen = ::recvfrom(udpchecksocs[i], pmsg->mpbuf, MonUdpSocketInstance::UDPMSG_SIZE, 
                    0, (struct sockaddr*)&pmsg->maddr, &nmln);
				ins.vrecvbuf.push_back(pmsg);
			}
		}
	}

	void SocketMulplexService_Select::do_writable()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
            //M_DEBUG("check do_writable %d\n", tcpchecksocs[i]);
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], (fd_set*)pwritedfs))
			{
				if((ins.waitevt & EVT_CONNECTED) != 0x00)
					ins.socevent |= EVT_CONNECTED;
				else if(/*(ins.waitevt & EVT_WRITE_READY) != 0x00*/ ins.send_queue.size() > 0)
				{
                    M_DEBUG("writable %d\n",tcpchecksocs[i]);
					while ( ins.send_queue.size() > 0 )
					{
                        //M_DEBUG("ins.send_queue.size():%d\n", ins.send_queue.size());
						miniutil::data::shared_buffer b = ins.send_queue.front();
						int sendlen =  (int)b->length() > left_uquota?left_uquota:(int)b->length();
						if(sendlen <= 0)
                        {
                            M_DEBUG("no left quota\n");
							break;
                        }

                        //M_DEBUG("left quota:%d\n", left_uquota);
                        //int wantsendlen = b->length();
                        //if(wantsendlen > 512)
                        //    wantsendlen = 512;

						int len = ::send( tcpchecksocs[i], b->data(), (int)b->length(), 0 );
                        M_DEBUG("::send len %d\n", len);
						if ( len == (int)b->length() )
						{
							//M_DEBUG("do_send( )len %d == length()%d\n", len, (int)b->length());
							ins.send_queue.pop_front();
							left_uquota -= len;
							if(left_uquota <= 0)
                            {
                                M_DEBUG("No UP quota available:%d %d\n", left_uquota, len);
								break;
                            }
                            M_DEBUG("do_send( )len %d = length()%d\n", len, (int)b->length());
							//continue;
						}
						if ( len < 0)
						{
							int ret = miniutil::socket::getLastError();
							if(ret != EWOULDBLOCK)
                            {
                                ins.sendret = -1;
								ins.socevent |= EVT_WRITE_READY;
                                M_DEBUG("send_len < 0 ret:%d, err:%s\n", ret, miniutil::socketcmn::get_soc_errorstr(ret).c_str());
                            }
						}
						else if ( len > 0 && len <  (int)b->length() )
						{
							M_DEBUG("do_send( )len %d<length()%d\n", len, (int)b->length());
							b->erase( 0 , (size_t)len );
							left_uquota -= len;
						}
						else if(len == 0)
						{
							ins.socevent |= EVT_WRITE_READY;
						}
						break;
					}
				}
			}
		}
		for(unsigned int i=0; i< udpchecksocs.size(); i++)
		{
			socklen_t nmln = sizeof(struct sockaddr);
			MonUdpSocketInstance &ins = malludpsockets[udpchecksocs[i]];
			if(FD_ISSET(udpchecksocs[i], (fd_set*)pwritedfs))
			{
				if(ins.vrecvbuf.size() > 0)
				{
					MonUdpSocketInstance::UDPMSG* pmsg = ins.vrecvbuf[0];
					::sendto(ins.soc, pmsg->mpbuf, pmsg->mlen, 0, (struct sockaddr*)&pmsg->maddr, nmln);
					delete pmsg;
					ins.vrecvbuf.erase(ins.vrecvbuf.begin());
				}
			}
		}
	}

	void SocketMulplexService_Select::do_exception()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], (fd_set*)pexcepdfs))
				ins.socevent |= EVT_EXCEPTION;
		}
	}

} }
